package kmeans;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.*;
import java.util.ArrayList;

public class KMeansDriver {
    private final int k;
    private int iterationNum;
    private final String sourcePath;
    private final String outputPath;
    private final Configuration configuration;
    private final FileSystem fileSystem;

    public KMeansDriver(int k, int iterationNum, String sourcePath, String outputPath) throws IOException {
        this.k = k;
        this.iterationNum = iterationNum;
        this.sourcePath = sourcePath;
        this.outputPath = outputPath;
        this.configuration = new Configuration();
        fileSystem = FileSystem.get(this.configuration);
        System.out.println(k + "," + iterationNum + "," + sourcePath + "," + outputPath);
    }

    protected ArrayList<Cluster> setup(String ClusterPath) throws IOException {
        ArrayList<Cluster> kClusters = new ArrayList<Cluster>();
        System.out.println("judge " + ClusterPath);
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(ClusterPath));
        for (FileStatus fileStatus : fileStatuses) {
            if (!fileStatus.isDirectory()) {
                FSDataInputStream fsDataInputStream = fileSystem.open(fileStatus.getPath());
                BufferedReader in = new BufferedReader(new InputStreamReader(fsDataInputStream));
                String line = null;
                while ((line = in.readLine()) != null) {
                    Cluster cluster = new Cluster(line);
                    kClusters.add(cluster);
                }
                in.close();
                fsDataInputStream.close();
            }
        }
        return kClusters;
    }

    public boolean isFinished(ArrayList<Cluster> pre, ArrayList<Cluster> now) {
        if (pre.size() != now.size()) {
            try {
                throw new Exception("cluster number not compatible");
            } catch (Exception e) {
                e.printStackTrace();
                System.exit(-1);
            }
        }
        for (int i = 0; i < pre.size(); i++) {
            if (!pre.get(i).equals(now.get(i))) {
                return false;
            }
        }
        return true;
    }

    //KMeans的任务配置，迭代调用，直至收敛或者迭代次数达到上限
    public boolean KMeansJob() throws IOException, ClassNotFoundException, InterruptedException {
        for (int i = 1; i <= iterationNum; i++) {
            String prePath = outputPath + "/cluster-" + (i - 1) + "/";
            String newPath = outputPath + "/cluster-" + i + "/";
            if (i >= 2) {
                ArrayList<Cluster> preCluster = setup(prePath);
                ArrayList<Cluster> nowClusrer = setup(newPath);
                if (isFinished(preCluster, nowClusrer)) {
                    System.out.println("Done iterationNum: " + i);
                    iterationNum = i;
                    break;
                }
            }
            KMeans.run(i, sourcePath, outputPath + "/cluster-" + (i + 1) + "/",
                    outputPath + "/cluster-" + i + "/");
        }
        return true;
    }

    //KMeansCluster的任务配置
    public void KMeansClusterJob() throws IOException, ClassNotFoundException, InterruptedException {
        KMeansCluster.run(sourcePath, outputPath + "/clusteredInstances/",
                outputPath + "/cluster-" + iterationNum + "/");
        System.out.println("cluster finished");
    }

    //确定初始的k个中心点
    public boolean generateInitialCluster() throws IOException {
        BufferedReader bufferedReader = new BufferedReader(new FileReader(sourcePath));
        fileSystem.mkdirs(new Path(outputPath + "/cluster-1/"));

        FSDataOutputStream outputStream = fileSystem.create(new Path(outputPath + "/cluster-1/original-center.txt"));
        BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(outputStream));
        String line = "";
        for (int i = 0; i < k; i++) {
            if ((line = bufferedReader.readLine()) != null) {
                bufferedWriter.write(i + ",1," + line + "\t\n");
            }
        }
        bufferedReader.close();
        bufferedWriter.close();
        return true;
    }

    public void copyFile(String src, String dest) throws IOException {
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path(dest));
        BufferedWriter out = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream));
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(src));
        for (FileStatus fileStatus : fileStatuses) {
            if (!fileStatus.isDirectory()) {
                FSDataInputStream fsDataInputStream = fileSystem.open(fileStatus.getPath());
                BufferedReader in = new BufferedReader(new InputStreamReader(fsDataInputStream));
                String line = null;
                while ((line = in.readLine()) != null) {
                    out.write(line + "\t\n");
                }
                in.close();
                fsDataInputStream.close();
            }
        }
        out.close();
    }

    // 将结果输出至目标文件，并删除中间输出结果
    public void finishJob() throws IOException {
        moveResult();
        for (int i = 1; i <= iterationNum; i++) {
            deleteFile(outputPath + "/cluster-" + i);
        }
        deleteFile(outputPath + "/clusteredInstances");
    }

    public void moveResult() throws IOException {
        String ClusterCenterSrc = outputPath + "/cluster-" + iterationNum;
        String ClusterCenterDest = outputPath + "/cluster-centers";
        String ClusterPointsSrc = outputPath + "/clusteredInstances";
        String ClusterPointsDest = outputPath + "/cluster-points";
        copyFile(ClusterCenterSrc, ClusterCenterDest);
        copyFile(ClusterPointsSrc, ClusterPointsDest);
    }

    // 删除文件或者文件夹
    public void deleteFile(String path) throws IOException {
        fileSystem.deleteOnExit(new Path(path));
    }

    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
        KMeansDriver kMeansDriver = new KMeansDriver(
                Integer.parseInt(args[0]), Integer.parseInt(args[1]), args[2], args[3]);

        if (!kMeansDriver.generateInitialCluster()) {
            try {
                throw new Exception("generateInitialCluster failed");
            } catch (Exception e) {
                e.printStackTrace();
                System.exit(-1);
            }
        }
        if (kMeansDriver.KMeansJob()) {
            kMeansDriver.KMeansClusterJob();
            kMeansDriver.finishJob();
        }
    }
}
